#pragma once
#include <etcd/Client.hpp>
#include <etcd/Response.hpp>
#include <etcd/KeepAlive.hpp>
#include <etcd/Watcher.hpp>
#include "Logger.hpp"

namespace XKL
{
    class RpcRegister
    {
    public:
        using ptr = std::shared_ptr<RpcRegister>;
        RpcRegister(const std::string& register_host)
            : _register_host(register_host)
            , _etcd_client(new etcd::Client(_register_host))
            , _keepalive(_etcd_client->leasekeepalive(_lease_id_sec).get())
            , _lease_id(_keepalive->Lease())
        {}

        void onRegister(const std::string& service_name, const std::string& service_host)
        {
            auto resp = _etcd_client->put(service_name, service_host, _lease_id).get();
            if(!resp.is_ok())
                LOG_ERROR("{}-{}, service registration failed, reason: {}", service_name, service_host, resp.error_message());
            LOG_DEBUG("{}-{}, service registration successed", service_name, service_host);
        }

        void cancel()
        {
            _etcd_client->leaserevoke(_lease_id);
        }

        ~RpcRegister()
        {
            _etcd_client->leaserevoke(_lease_id);
            _keepalive->Cancel();
        }
    private:
        std::string _register_host;                 // 服务注册中心地址
        const size_t _lease_id_sec = 3;             // 过期时间
        std::unique_ptr<etcd::Client> _etcd_client; // 服务中心客户端
        std::shared_ptr<etcd::KeepAlive> _keepalive;// 租约对象
        int64_t _lease_id;                          // 租约id
    };

    class Discovery
    {
    public:
        using ptr = std::shared_ptr<Discovery>;
        using NotifyCallback = std::function<void(std::string, std::string)>;
        Discovery(const std::string &host, const std::string &base_dir,
            const NotifyCallback &put_cb, const NotifyCallback &del_cb)
            : _client(new etcd::Client(host)), _put_cb(put_cb), _del_cb(del_cb)
        {
            auto resp = _client->ls(base_dir).get();
            if(!resp.is_ok())
                LOG_ERROR("get service message failed!");
            int n = resp.keys().size();
            for(int i = 0; i < n; i++)
                if(_put_cb) _put_cb(resp.key(i), resp.value(i).as_string());
            _watcher = std::make_shared<etcd::Watcher>(*_client.get(), base_dir, std::bind(&Discovery::callback, this, std::placeholders::_1), true);
        }  
    private:
        void callback(const etcd::Response &resp)
        {
            if(!resp.is_ok())
            {
                LOG_ERROR("received an incorrect event notification");
                return;
            }
            for(const auto& ev : resp.events())
            {
                if(ev.event_type() == etcd::Event::EventType::PUT)
                {
                    if(_put_cb) _put_cb(ev.kv().key(), ev.kv().as_string());
                    LOG_DEBUG("new service: {}-{}", ev.kv().key(), ev.kv().as_string());
                }
                else if(ev.event_type() == etcd::Event::EventType::DELETE_)
                {
                    if(_del_cb) _del_cb(ev.prev_kv().key(), ev.prev_kv().as_string());
                    LOG_DEBUG("del service: {}-{}", ev.prev_kv().key(), ev.prev_kv().as_string());
                }
            }
        }
    private:
        NotifyCallback _put_cb;
        NotifyCallback _del_cb;
        std::shared_ptr<etcd::Client> _client;
        std::shared_ptr<etcd::Watcher> _watcher;
    };
}